package org.jetlinks.pro.messaging.kafka.impl;

import org.jetlinks.pro.datasource.CommandHandler;
import org.jetlinks.pro.datasource.DataSource;
import org.jetlinks.pro.messaging.kafka.*;
import org.jetlinks.pro.messaging.kafka.command.CreateConsumer;
import org.jetlinks.pro.messaging.kafka.command.CreateProducer;
import reactor.core.publisher.Mono;

class CreateProducerHandler implements CommandHandler<CreateProducer, Mono<KafkaProducer>> {

     static CreateProducerHandler INSTANCE = new CreateProducerHandler();

    @Override
    public Mono<KafkaProducer> handle(DataSource dataSource, CreateProducer command) {
        KafkaDataSource kafkaDataSource = dataSource.unwrap(KafkaDataSource.class);

        ReactorKafkaProducer consumer = new ReactorKafkaProducer(command.getProperties() == null
                                                              ? kafkaDataSource.getConfig()
                                                              : command.getProperties());


        return Mono.just(consumer);
    }
}
